package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.DnsCount;
import com.surfilter.massdata.spark.bean.SA7601;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.DateUtil;


public class DomainAnalyzTask_SA7601ZT_dayNew2 implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA7601ZT_dayNew2.class);
	private static final long serialVersionUID = 1L;
	private static String outName;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		long temp=start;
		long temp_end=start;
		double exectime=0.0;
		
		
		System.out.println("==============DomainAnalyzTask_SA7601 begin====================");
		System.out.println("............本省主体....开始............");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			String p_linces_key = new String(ConfigUtil.get("LICENSE_KEY").getBytes("ISO-8859-1"),"UTF-8");
			System.out.println("LICENSE_KEY="+p_linces_key);
			DataFrame dataSource = dataFrames.get("dws_hour_sa7601");
			String ztsql="select domain,sum(traffic)as count  from dws_hour_sa7601_temp where  buss_type='WEBSITE_RANK_SUM' and sta_range=2 and BUSS_VALUE=0 GROUP BY domain";
			DataFrame dfzt_sa7601=dataSource.sqlContext().sql(ztsql);
			
			String dayStr = commandMap.get("-d");
		    Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);

			JavaRDD<Map<String, String>> domainPair = dfzt_sa7601.toJavaRDD().map(new Function<Row,Map<String,String>>(){
				private static final long serialVersionUID = 1L;
					@Override
					public Map<String, String> call(Row row) throws Exception {
						Map<String, String> map=new HashMap<String,String>();
						String domain=row.getAs("domain").toString();
						String count=row.getAs("count").toString();
						map.put("domain", domain);
						map.put("count", count);
						System.out.println("主体map:"+map);
						return map;
					}
					
				});
  
	        @SuppressWarnings("serial")
			JavaRDD<DnsCount> dnsRDD = domainPair.mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>,DnsCount>() {

				@Override
				public Iterable<DnsCount> call(Iterator<Map<String, String>> it){
					List<DnsCount> list = new ArrayList<DnsCount>(); 
					while(it.hasNext()){
						 Map<String, String> map= it.next();
						String top_domain =map.get("domain").toString();
						long count=Long.parseLong(map.get("count").toString());
						if(StringUtils.isNotBlank(top_domain)){
							DnsCount sa = new DnsCount();
							sa.setTop_domain(top_domain);
							sa.setCount(count);
							list.add(sa);
							System.out.println("主体map:"+list);
						}
					}
					return list;
				}
			});
	        
			DataFrame df_dc = dataSource.sqlContext().createDataFrame(dnsRDD, DnsCount.class);
	        df_dc.registerTempTable("df_dc_temp");
	        
	        String sql_beian = "select a.top_domain,a.count, "
					+ "(CASE WHEN ISNULL(b.license_key) = false THEN B.license_key ELSE 'N/A' END) AS license_key,"
					+ "(CASE WHEN ISNULL(b.site_name) = false THEN B.site_name ELSE 'N/A' END) AS site_name "
					+ "from df_dc_temp a left join br2005_temp b on a.top_domain = b.top_domain where b.LICENSE_KEY like '%"+p_linces_key+"%' order by a.count desc "; 

	        DataFrame oper_tmp_beian = dataSource.sqlContext().sql(sql_beian);
	        
			List<Row> top_domain_100_beian = oper_tmp_beian.toJavaRDD().take(100);
			List<SA7601> list_SA7601 = new ArrayList<SA7601>();	
			if(top_domain_100_beian.size() > 0){
	    		int rank = 1;
	        	for(Row row_top_domain_100_beian : top_domain_100_beian){
	        	    SA7601 sa = new SA7601();
	        		sa.setYear(DateUtil.getCurrentYear(date));
	        		sa.setHalf_year(DateUtil.getHalfYear(date));
	        		sa.setQuarter(DateUtil.getQuarter(date));
	        		sa.setMonth(DateUtil.getCurrentMonth(date));
	        		sa.setWeek(DateUtil.getCurrentWeek(date));
	        		sa.setDay(DateUtil.getCurrentDay(date));
	        		sa.setBuss_type("WEBSITE_RANK_SUM");
	        		sa.setBuss_value("0");
	        		sa.setRank(rank++);
	        		sa.setDomain(row_top_domain_100_beian.getAs("top_domain").toString());
	        	    sa.setWebsite_name(row_top_domain_100_beian.getAs("site_name").toString());
	        	    sa.setIcp_num(row_top_domain_100_beian.getAs("license_key").toString());
	        	    sa.setTraffic(row_top_domain_100_beian.getAs("count").toString());
	        		sa.setSta_range(2); //本省主体是2
	        		list_SA7601.add(sa);
	        	}
	        }
	        
	        df_dc.sqlContext().dropTempTable("df_dc_temp");
	        df_dc.unpersist();
	  
			JavaSparkContext ctx = new JavaSparkContext(dataSource.sqlContext().sparkContext());
			//HiveContext hiveContext = new HiveContext(ctx);
			//hiveContext.dropTempTable("df_dc_temp");
	    	JavaRDD<SA7601> rdd = ctx.parallelize(list_SA7601);
	    	DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SA7601.class);
			outList.add(new OutQueueEntity(outName, df));
			
		
			CommonUtils.deleteTaskTableData("SA7601", date, "WEBSITE_RANK_SUM", 2, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("DomainAnalyzTask_SA7601_New本省主体:exectime: "+min+" min............");
		return outList;
	}
}
